#include "mpimapreduce.h"
#include <iostream>
using namespace std;



class MyScheduler:public MapReduceScheduler<int,int,int,int> {
public:
	MyScheduler(int rank,int procs):MapReduceScheduler<int,int,int,int>(rank,procs){}

	virtual void map(const int*data, const unsigned int len){
		int sum = 0;
		for(unsigned int i =0;i<len;i++){
			sum += data[i];
		}
		emit_intermediate(1,sum);
	}
	
	virtual void reduce(const int&, const MapOutputValIter& valBegin, const MapOutputValIter& valEnd){
		int sum = 0;
		for(MapOutputValIter it = valBegin; it != valEnd; it++){
			sum+=*it;
		}	
		emit(sum);
	}
};


const int N = 10;
const int MODE = 100;

void gen_ints(int *arr,int n){
	srand(1000);
	for(int i =0; i<n ;i++){
		arr[i] = rand()%MODE;
	}
}

void print_ints(int *arr,int n){
	for(int i=0;i<n;i++){
		cout<<arr[i]<<" ";
	}
	cout<<endl;
}

int main(int argc,char** argv){
	int rank;
	int numprocs;
	MPI_Init(&argc,&argv);
	MPI_Comm_rank(MPI_COMM_WORLD,&rank);
	MPI_Comm_size(MPI_COMM_WORLD,&numprocs);
	
	MyScheduler my(rank,numprocs);

	if(rank == 0){
		int* ints = new int[N];
		gen_ints(ints,N);
		print_ints(ints,N);
		my.set_inputs(ints,N);
	}

	my.run();

	if(rank == 0){
		const map<int,int>& output =  my.get_outputs();
		for(map<int,int>::const_iterator iter = output.begin();iter!= output.end();iter++){
			cout<<"("<<iter->first<<","<<iter->second<<")"<<endl;
		}
	}

	MPI_Finalize();
}

